package com.study.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.RoundRobinAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

/**
 * @author
 * @date 2022年09月10日 21:06
 * @description
 */
public class CustomConsumer2 {


    public static void main(String[] args) {


        // 配置

        Properties properties = new Properties();


        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.160:9092");


        // 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 指定group id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

        // 指定分区策略
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());


        // 创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 指定主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");

        // 订阅主题
        kafkaConsumer.subscribe(topics);

        // 消费数据
        while (true) {
            for (ConsumerRecord<String, String> consumerRecord : kafkaConsumer.poll(Duration.ofSeconds(1))) {
                System.out.println(consumerRecord);
            }
        }


    }
}
